package com.gupao.javavip.dubbo.provider;

import com.alibaba.nacos.api.config.annotation.NacosValue;
import lombok.extern.apachecommons.CommonsLog;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

@CommonsLog
@Component
@EnableScheduling
public class KafkaMsgProducer implements ApplicationListener<ContextRefreshedEvent> {

    @SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
    @Autowired
    private KafkaTemplate<String, String> template;

    @NacosValue(value = "${biz.send-msg-count:1}", autoRefreshed = true)
    private int sendMsgCount;

    @Override
    public void onApplicationEvent(ContextRefreshedEvent event) {
        sendMsg();
    }

    @Scheduled(cron = " 0/10 * * * * ? ")
    public void sendMsg() {
        String topic = "test";
        for (int i = 0; i < sendMsgCount; i++) {
            String msg = "MSG-" + i + "-" + System.currentTimeMillis();
            ListenableFuture<SendResult<String, String>> future = template.send(topic, msg, msg);
            future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
                @Override
                public void onFailure(Throwable ex) {
                    log.error("kafka sendMessage error, [" + topic + "," + msg + "]," + ex.getMessage(), ex);
                }

                @Override
                public void onSuccess(SendResult<String, String> result) {
                    log.info("kafka sendMessage success [" + topic + "," + msg + "]");
                }
            });
            log.debug("kafka sendMessage end");
        }
    }
}
